// 
// Copyright (C) 2000-2003
// magustek co.
//

#ifndef	MAGUS_SERVER_HPP
#define	MAGUS_SERVER_HPP

#include <magus/config.hpp>		
#include <magus/utility.hpp>		
#include <magus/network/ServerSocket.hpp>
#include <magus/network/Socket.hpp>
#include <magus/thread/thread.hpp>
#include <magus/thread/mutex.hpp>
#include <magus/thread/condition.hpp>
#include <queue>
#include <list>
#include <cassert>
#include <cstdio>

#include <magus/debug_tools/debug_log.hpp>

using magus::debug_output;

#include <iostream>
using namespace std;

namespace magus {

struct server_param 
{
	int port;
	int timeout;	// in seconds
	int	num_threads;
	int listen_signal;
};

template <class workerT>
class server : private noncopyable, public thread_functor
{
public:
	int serial;
	// Constructor & destructor
	explicit server(server_param p) 
		: _M_running(false), _M_exiting(false), _M_param(p), _M_listener(NULL)
	{
		serial = 0;
	}
	
	virtual ~server() 
	{
        this->close();
	}

	// open server, begin to listen
	bool open()
	{
		// server already running
		if (_M_listener != NULL)
			return true;

		_M_exiting = false;
		
		// Create server socket
		int rv = _M_svr_socket.Create(_M_param.port);
		if (rv == -1) {
			return false;
		}
		_M_listener = new thread(*this);
		if (_M_listener == NULL) {
			_M_svr_socket.Close();
			return false;
		}
		return true;
	}

	// close server, stop listen
	void close()
	{
		_M_exiting = true;
		if (_M_listener != NULL)
		{
			interrupt_server();
			_M_listener->join();
			delete _M_listener;
			_M_listener = NULL;
		}
	}
	
	bool getTask(socket_t& task_)
	{
		if (_M_exiting)
			return false;
		// wait until queue isn't empty
		mutex::scoped_lock lock(_M_mtx);
		while (_M_queue.empty()) 
		{
			_M_not_empty.wait(lock);
			if (_M_exiting)
				return false;
		}
		assert(!_M_queue.empty());
		
		// get task
		task_ = _M_queue.front();
		_M_queue.pop();
		return true;
	}

	bool wait_for_idle_worker()
	{
		mutex::scoped_lock lock(_M_mtx);
		while (_M_idle_workers == 0) 
		{
			_M_idle_condition.wait(lock);
			if (_M_exiting) {
				return false;
			}
		}
		return true;
	}

	void release(int worker_id)
	{
		mutex::scoped_lock lock(_M_mtx);
		_M_idle_workers++;
		_M_idle_condition.notify_all();
	}

	// is server running ?
	bool is_running() { return _M_running; }

	// is server running ?
	bool is_exiting() { return _M_exiting; }

	// listener thread function
	void operator()(void)
	{
		char msg[128];
		debug_output("listener started");
		_M_running = true;
		for (int i=0; i<_M_param.num_threads; ++i)
		{
			workerT* w = new workerT(*this, i);
			if (w != NULL) {
				_M_workers.push_back(w);
				_M_worker_group.create_thread(*w);
			}
		}
		_M_idle_workers = _M_worker_group.thread_size();
		if (_M_idle_workers == 0) {
			_M_running = false;
			cout << "listener exit\n";
			return;
		}
		sprintf(msg, "number of workers created %d", _M_idle_workers);
		debug_output(msg);
		while (!_M_exiting)
		{
			//if (!wait_for_idle_worker())
				//continue;
			socket_t s;
			int rv = _M_svr_socket.Accept(s);
			if (rv != 0) {
				sprintf(msg, "accept error %d", rv);
				debug_output(msg);
				continue;
			}
			if (_M_idle_workers <= 0)
			{
				debug_output("connection is too large, please wait");
				Socket sock(s);
				sock.Close();
				continue;
			}

			sprintf(msg, "number of idle workers %d", _M_idle_workers);
			debug_output(msg);

			s.timeout = _M_param.timeout;
			assert(rv == 0 || s.fd != INVALID_SOCKET);
			
			mutex::scoped_lock lock(_M_mtx);
			_M_queue.push(s);
			_M_idle_workers--;
			_M_not_empty.notify_one();
		}
		// cleanup
		_M_svr_socket.Close();	

		// wake up all workers
		_M_not_empty.notify_all();
		_M_worker_group.join_all();
		for (typename std::list<workerT*>::iterator it = _M_workers.begin();
			it != _M_workers.end(); ++it) 
		{
			delete (*it);
		}
		_M_workers.clear();
		debug_output("listener exit");
		_M_running = false;
	}

private:
	bool 				_M_running;			// server running flag
	bool 				_M_exiting;			// server exiting flag
	server_param 		_M_param;			// 
	ServerSocket 		_M_svr_socket;		// server socket
	thread* 			_M_listener;		// pointer to task process thread
	std::list<workerT*>	_M_workers;			// worker list
	thread_group		_M_worker_group;	//
	mutex 				_M_mtx;				// lock for task queue sync
	condition 			_M_not_empty;		// task queue not empty condition
	int 				_M_idle_workers;	// idle worker count
	condition 			_M_idle_condition;	// idle worker available
	std::queue<socket_t> _M_queue;			// task queue wait for process
	
	// wake up listen thread
	void interrupt_server()
	{
		// notify all worker
		_M_not_empty.notify_all();
		
		// notify listener
		_M_idle_condition.notify_all();
	#ifdef __MAGUS_POSIX    
		_M_listener->interrupt(_M_param.listen_signal);
	#else
		_M_svr_socket.Close();	
	#endif    
	}

}; // class server

} // namespace magus

// Change Log:
//    02 JLY 2003  Initial version.

#endif	/* MAGUS_SERVER_HPP */
